package com.freez.spark.core.operation;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: OtherOperation.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:12
 * Description: 对intersection,union,subtract和subtractByKey,</br>
 *              cartesian,join,cogroup,leftOuterJoin,rightOuterJoin</br>
 *              算子进行演示
 */
public class OtherOperation {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("SparkCore-OtherOperationRDD").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<Integer> oneVRDD = jsc.parallelize(Arrays.asList(1, 3, 6, 7, 5));
        JavaRDD<Integer> twoVRDD = jsc.parallelize(Arrays.asList(2, 3, 6, 4, 0));
        JavaPairRDD<String, Integer> oneKVRDD = jsc.parallelizePairs(Arrays.asList(
                new Tuple2<>("张三", 20),
                new Tuple2<>("李华", 40),
                new Tuple2<>("王强", 60),
                new Tuple2<>("秦泽", 70)
//                new Tuple2<>("钱芳", 88),
//                new Tuple2<>("牟可", 88)
        ));
        JavaPairRDD<String, Integer> twoKVRDD = jsc.parallelizePairs(Arrays.asList(
                new Tuple2<>("孟明", 22),
                new Tuple2<>("李华", 40),
                new Tuple2<>("王强", 64),
                new Tuple2<>("刘鑫", 77)
//                new Tuple2<>("钱芳", 80)
        ));
        /**
         * intersection交集
         */
        /*oneVRDD.intersection(twoVRDD).foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer s) throws Exception {
                System.out.println("intersection>>>>>>V>>>>>" + s);
            }
        });
        oneKVRDD.intersection(twoKVRDD).foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> s) throws Exception {
                System.out.println("intersection>>>>>KV>>>>>" + s);
            }
        });*/
        /**
         * union并集
         */
        /*JavaPairRDD<String, Integer> oneRDD = oneKVRDD.repartition(4);
        JavaPairRDD<String, Integer> twoRDD = twoKVRDD.repartition(3);
        JavaPairRDD<String, Integer> unionKVRDD = oneRDD.union(twoRDD);
        System.out.println("union之后的分区数：" + unionKVRDD.getNumPartitions());
        System.out.println(unionKVRDD.collect());
        unionKVRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> s) throws Exception {
                System.out.println("union>>>>>KV>>>>>" + s);
            }
        });

        oneVRDD.union(twoVRDD).foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer s) throws Exception {
                System.out.println("union>>>>>V>>>>>" + s);
            }
        });*/
        /**
         *  subtract和subtractByKey  差集
         */
//        JavaPairRDD<String, Integer> oneRDD = oneKVRDD.repartition(5);
//        JavaPairRDD<String, Integer> twoRDD = twoKVRDD.repartition(3);
//        JavaPairRDD<String, Integer> subtractKVRDD = oneRDD.subtract(twoRDD);
//        JavaPairRDD<String, Integer> subtractByKeyKVRDD = oneRDD.subtractByKey(twoRDD, 2);
//        System.out.println("subtractKVRDD>>>>>>" + subtractKVRDD.collect() + "，分区个数=" + subtractKVRDD.getNumPartitions());
//        System.out.println("subtractByKeyKVRDD>>>>>>" + subtractByKeyKVRDD.collect() + "，分区个数=" + subtractByKeyKVRDD.getNumPartitions());
        /**
         * cartesian 笛卡尔集
         */
        /*JavaPairRDD<Tuple2<String, Integer>, Tuple2<String, Integer>> cartesianRDD = oneKVRDD.cartesian(twoKVRDD);
        cartesianRDD.foreach(new VoidFunction<Tuple2<Tuple2<String, Integer>, Tuple2<String, Integer>>>() {
            @Override
            public void call(Tuple2<Tuple2<String, Integer>, Tuple2<String, Integer>> s) throws Exception {
                System.out.println(s);
            }
        });
        System.out.println("分区个数=" + cartesianRDD.getNumPartitions());
*/
        /**
         * join 和 cogroup
         */
//        oneKVRDD.join(twoKVRDD).foreach(new VoidFunction<Tuple2<String, Tuple2<Integer, Integer>>>() {
//            @Override
//            public void call(Tuple2<String, Tuple2<Integer, Integer>> s) throws Exception {
//                System.out.println("join >>> " + s );
//            }
//        });
//        oneKVRDD.cogroup(twoKVRDD).foreach(new VoidFunction<Tuple2<String, Tuple2<Iterable<Integer>, Iterable<Integer>>>>() {
//            @Override
//            public void call(Tuple2<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> s) throws Exception {
//                System.out.println("cogroup >>> " + s);
//            }
//        });
        /**
         * leftOuterJoin
         */
//        oneKVRDD.leftOuterJoin(twoKVRDD).foreach(new VoidFunction<Tuple2<String, Tuple2<Integer, Optional<Integer>>>>() {
//            @Override
//            public void call(Tuple2<String, Tuple2<Integer, Optional<Integer>>> s) throws Exception {
//                System.out.println(s);
//            }
//        });
        /**
         * rightOuterJoin
         */
        /*JavaPairRDD<String, Tuple2<Optional<Integer>, Integer>> s = oneKVRDD.rightOuterJoin(twoKVRDD);
        s.foreach(new VoidFunction<Tuple2<String, Tuple2<Optional<Integer>, Integer>>>() {
            @Override
            public void call(Tuple2<String, Tuple2<Optional<Integer>, Integer>> s) throws Exception {
                System.out.println(s);
            }
        });*/
        oneKVRDD.join(oneKVRDD).foreach(new VoidFunction<Tuple2<String, Tuple2<Integer, Integer>>>() {
            @Override
            public void call(Tuple2<String, Tuple2<Integer, Integer>> s) throws Exception {
                System.out.println(s);
            }
        });

        jsc.close();
    }
}